package com.ruoyi.vehicle.netty;

import cn.hutool.core.util.RandomUtil;
import com.ruoyi.vehicle.entity.MsgDecoder;
import com.ruoyi.vehicle.entity.MsgEncoder;
import com.ruoyi.vehicle.entity.PackageData;
import com.ruoyi.vehicle.entity.request.AuthMsg;
import com.ruoyi.vehicle.entity.request.HeartMsg;
import com.ruoyi.vehicle.entity.request.LocationMsg;
import com.ruoyi.vehicle.entity.request.RegisterMsg;
import com.ruoyi.vehicle.entity.response.CommonMsgRespBody;
import com.ruoyi.vehicle.entity.response.RegisterMsgRespBody;
import com.ruoyi.vehicle.utils.Constants;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author : CZQ
 * @date : 2024-09-17 15:57
 * @description :
 **/
@Slf4j
@Component
@ChannelHandler.Sharable
public class NettyHandler extends ChannelInboundHandlerAdapter {

    private MsgDecoder decoder = new MsgDecoder();

    private MsgEncoder encoder = new MsgEncoder();

    // 用于保存连接对象，以连接ID为key， 连接对象为value
    private static Map<String, ChannelHandlerContext> socketMap = new ConcurrentHashMap();


    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    //https://gitcode.com/gh_mirrors/jt/jt-808-protocol/blob/master/jt808-tcp-netty/src/main/java/cn/hylexus/jt808/service/BaseMsgProcessService.java

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // (2)
        // 获取连接ID，每一次连接要求消息流水号从0开始连续
        String shortText = ctx.channel().id().asShortText();
        int flowId = stringRedisTemplate.opsForValue().increment("socket:"+shortText).intValue()-1;
        stringRedisTemplate.expire("socket:"+shortText, 15, TimeUnit.MINUTES);

        try {
            ByteBuf buf = (ByteBuf) msg;
            if (buf.readableBytes() <= 0) {
                return;
            }
            // 读取收到的数据，转成byte[]
            byte[] bs = new byte[buf.readableBytes()];
            buf.readBytes(bs);
            PackageData jt808Msg = this.decoder.queueElement2PackageData(bs);

            // 处理客户端消息,得到需要返回的报文byte[]
            byte[] bytes = this.processClientMsg(jt808Msg, ctx, flowId);

            if(bytes != null){
                //发送给终端
                ChannelFuture future = ctx.channel().writeAndFlush(Unpooled.copiedBuffer(bytes)).sync();
                if (!future.isSuccess()) {
                    log.error("发送数据出错:{}", future.cause());
                }
            }
        } finally {
            release(msg);
        }

    }

    private byte[] processClientMsg(PackageData jt808Msg, ChannelHandlerContext ctx, int flowId) throws Exception {
        // 根据消息ID的不同,分别实现自己的业务逻辑
        if(jt808Msg.getMsgHeader().getMsgId() == Constants.msg_id_terminal_location_info_upload){
            LocationMsg msg = this.decoder.toLocationMsg(jt808Msg);
            log.info("收到[位置信息汇报]请求：{}",msg.getLocationInfo());

            // TODO 将车辆最新的经纬度存入Redis


            // TODO 将解析出的车辆数据msg发送到Kafka异步处理（需要处理报警数据和车辆状态数据），在这里不做处理


            //  kafka消费之后再入库，应避免消息丢失和重复消费（Kafka消费之后再存入Hbase）
            //  kafka消费后。处理报警信息


            return this.getCommonResp(jt808Msg, msg, flowId, CommonMsgRespBody.success);
        }  else if(jt808Msg.getMsgHeader().getMsgId() == Constants.msg_id_terminal_heart_beat){
            log.info("收到[心跳]请求,无数据");
            HeartMsg msg = this.decoder.toHeartMsg(jt808Msg);

            // TODO 心跳10秒钟一次，每次收到心跳请求对redis中的车辆缓存数据进行续期，续期时间需要大于10秒，避免因为网络问题漏掉部分心跳数据


            return this.getCommonResp(jt808Msg, msg,flowId, CommonMsgRespBody.success);
        } else if(jt808Msg.getMsgHeader().getMsgId() == Constants.msg_id_terminal_authentication){
            AuthMsg msg = this.decoder.toAuthMsg(jt808Msg);
            log.info("收到[终端鉴权]请求,数据：{}",msg.getAuthInfo());
            // TODO 根据连接ID（ctx.channel().id().asShortText()）为key查询出正确鉴权码

            // TODO 根据msg中解析出的鉴权报文的鉴权码和上一步查出的鉴权码判断是否鉴权成功

            // TODO 如果鉴权成功，以连接ID（ctx.channel().id().asShortText()）为key，存入Redis（鉴权成功的标识）   用于延迟消息的校验
            // TODO 根据msg中解析出的车辆信息，更改车辆状态为上线

            // TODO 如果鉴权失败主动断开和设备的连接

            return this.getCommonResp(jt808Msg, msg, flowId, CommonMsgRespBody.success);
        } else if (jt808Msg.getMsgHeader().getMsgId() == Constants.msg_id_terminal_register) {
            // 解析请求报文，得到解析结果的实体类
            RegisterMsg msg = this.decoder.toRegisterMsg(jt808Msg);
            log.info("收到[终端注册]请求,数据：{}",msg.getRegisterInfo());

            // TODO 根据msg中解析出的车辆信息，判断车辆数据是否存在，不存在存入数据库

            // TODO 以连接ID（ctx.channel().id().asShortText()）为key（需要加前缀区分不同业务），车辆信息为值存入Redis,用于当断开连接时根据断开的连接ID更改车辆状态

            // 完成终端注册应答
            RegisterMsgRespBody respMsgBody = new RegisterMsgRespBody();
            respMsgBody.setReplyCode(RegisterMsgRespBody.success);
            respMsgBody.setReplyFlowId(jt808Msg.getMsgHeader().getFlowId());
            // 生成6位随机鉴权码
            String token = RandomUtil.randomString(6);
            respMsgBody.setReplyToken(token);

            // TODO 将鉴权码存入Redis用于鉴权报文的校验，以连接ID（ctx.channel().id().asShortText()）为key（需要加前缀区分不同业务），鉴权码为值

            return this.encoder.encodeRegisterResp(msg, respMsgBody, flowId);
        } else {
            log.error("未知消息,消息ID={}", jt808Msg.getMsgHeader().getMsgId());
            return null;
        }
    }

    /**
     * 生成通用应答数据
     * @param jt808Msg 用于获取应答流水号和消息ID
     * @param msg 用于生成应答的报文体
     * @param flowId 用于应答报文头中的消息流水号
     * @param result 返回的结果：成功/失败原因
     * @return 应答的byte[]
     * @throws Exception
     */
    public byte[] getCommonResp(PackageData jt808Msg, PackageData msg, int flowId, byte result) throws Exception {
        // 完成通用应答
        CommonMsgRespBody commonMsgRespBody = new CommonMsgRespBody();
        commonMsgRespBody.setReplyFlowId(jt808Msg.getMsgHeader().getFlowId());
        commonMsgRespBody.setReplyId(jt808Msg.getMsgHeader().getMsgId());
        commonMsgRespBody.setReplyCode(result);
        return this.encoder.encodeCommmonResp(msg, commonMsgRespBody, flowId);
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
        System.out.println("抓到异常");
        // TODO 根据连接ID（ctx.channel().id().asShortText()）查询出发成异常的车辆信息，做好日志记录
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("建立连接");

        socketMap.put(ctx.channel().id().asShortText(), ctx);
        // TODO 创建延迟消息，将连接ID（ctx.channel().id().asShortText()）发送到延迟队列，1分钟后检测是否鉴权成功，未鉴权成功主动断开连接

    }

    public static void closeSocket(String channelId){
        ChannelHandlerContext ctx = socketMap.get(channelId);
        if(ctx != null){
            ctx.channel().close();
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("断开连接");

        socketMap.remove(ctx.channel().id().asShortText());
        // TODO 根据连接ID（ctx.channel().id().asShortText()）查询Redis，获取连接对应的车辆信息，更改为下线

    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        // TODO 完成Netty自身的心跳机制
        // https://blog.csdn.net/qq_38887189/article/details/127262185
    }

    private void release(Object msg) {
        try {
            ReferenceCountUtil.release(msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
